/**
 * Created by emily on 17-11-15.
 */

import org.apache.kafka.clients.producer.*;
import kafka.producer.KeyedMessage;
import org.apache.kafka.common.protocol.types.Field;

import java.util.Properties;
import java.util.ArrayList;
import java.util.List;

public class KafkaBasic {

    // 声明静态配置
    static Properties props = new Properties();

    static {
        props.put("bootstrap.servers", "hadoop7:9092");
        props.put("metadata.broker.list", "hadoop7:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    }

    public void sendMessage(String topic, ArrayList<Messg> messgs){
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for (Messg messg: messgs){
            String m = messg.sensorId + "," + messg.value + "," + messg.timestamp;
            System.out.println("send to kafka: " + m);
            producer.send(new ProducerRecord<String, String>(topic, "5AB001-DX", m));
        }
        producer.close();
    }





}
